package org.scala.test.pool

import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.LinkedBlockingDeque
import scala.util.control.NonFatal

abstract class EventLoop[T](name:String) {
  
  protected val eventQueue:BlockingQueue[T] = new LinkedBlockingQueue[T]() 
  
  private val  stoped = new AtomicBoolean(false) 
  
  private val eventThread = new Thread(name){
    
     override def run():Unit={
         try{
          
            while(!stoped.get){
              val event = eventQueue.take()
              
               try{
                  onReceive(event)
               }catch{
                 case NonFatal(e)=>println("error threadName:"+name,e)
                 onError(e)
               }
            }
         }catch{
           case e:Throwable => e.printStackTrace()
           case NonFatal(e)=>println("error threadName:"+name,e)
         }
     }
  }
  
  
   def onStart(threadName:String):Unit={}
   
   def start():Unit={
     if(stoped.get)
        throw new IllegalStateException(name+"has bean stoped")
     
     onStart(name)
      eventThread.start()
   }
   
   def stop():Unit={
      if(stoped.compareAndSet(false, true)){
        eventThread.interrupt()
        eventThread.join() 
      }
   }
   
   protected def onReceive:PartialFunction[T,Unit] 
   
   protected def onError(e:Throwable):Unit 
   
}